package com.atguigu.state;

import com.atguigu.pojo.WaterSensor;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 *
 * @date 2024/11/20 15:40
 *
 * 案例需求：检测每种传感器的水位值，如果连续的两个水位值相差超过10，就输出报警。
 *
 * 思路：取出状态数据，与传进来值比较，更新状态数据，输出状态数据
 *
 * 按键分区状态 - 值状态 - ValueState
 *
 * ValueState: 单值状态， 状态中只会维护一个值
 *
 *     T value():  获取状态中的值
 *     void update(T value): 更新状态中的值
 *     void clear(): 清空状态
 *
 */
public class Flink05_KeyedValueState {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //数据源WaterSensor水位线类型：s1,100,1000
        SingleOutputStreamOperator<WaterSensor> ds = env.socketTextStream("hadoop102", 9999)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new WaterSensor(fields[0].trim(), Integer.valueOf(fields[1].trim()), Long.valueOf(fields[2].trim()));
                        }
                );

        ds.print("INPUT");

        ds
                .keyBy(WaterSensor::getId)
                .process(
                        new KeyedProcessFunction<String, WaterSensor, String>() {

                            //1.声明状态
                            private ValueState<Integer> lastVcState ;

                            //2.使用生命周期open方法，初始化状态

                            @Override
                            public void open(Configuration parameters) throws Exception {
                                RuntimeContext runtimeContext = getRuntimeContext();
                                ValueStateDescriptor<Integer> lastVcStateDesc = new ValueStateDescriptor<>("lastVcStateDesc", Types.INT);//todo 描述
                                lastVcState = runtimeContext.getState(lastVcStateDesc); //为什么把描述传过来？

                            }

                            @Override
                            public void processElement(WaterSensor waterSensor, KeyedProcessFunction<String, WaterSensor, String>.Context context, Collector<String> out) throws Exception {
                                //3.使用状态
                                //从状态中获取上一次的水为值
                                Integer lastValue = lastVcState.value();

                                //判断上一次水位值和这次的差值
                                if (lastVcState !=null && Math.abs(lastValue - waterSensor.getVc()) > 10){
                                    //更新状态
                                    lastVcState.update(waterSensor.getVc());
                                }
                            }
                        }
                )
                .print("PROCESS");




        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
